package com.itunion.quickstart.stream;

import com.itunion.quickstart.pojo.Student;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 *
 * 功能描述: 编写第一个HelloWorld，实时计算每个学生的最高分数
 *
 * @auther: lin
 * @date: 2021/9/23 10:32 上午
 */
public class HelloWorldStudent {
    public static void main(String[] args) throws Exception {
        // TODO 第一步：创建流计算执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // TODO 第二步：实时接收输入数据
        DataStream<String> inputDataStream = env.socketTextStream("localhost",8888);
        // TODO 第三步：转换数据类型
        DataStream<Student> dataStream = inputDataStream.map(new MapFunction<String, Student>() {
            @Override
            public Student map(String s) throws Exception {
                // TODO 按照逗号进行分割
                String[] fields = s.split(",");
                return new Student(fields[0],Double.parseDouble(fields[1]));
            }
        });
        // TODO 第四步：根据姓名进行分组
        KeyedStream<Student,String> keyedStream = dataStream.keyBy(student -> student.getName());
        // TODO 第五步：滚动聚合，计算每个学生的最高成绩
        DataStream<Student> resultStream = keyedStream.max("score");
        // TODO 第六步：打印输出
        resultStream.print();
        // TODO 第七步：执行
        env.execute();
    }
}
